package com.gome.han.bigdata.spark.core.rdd.operation.transformation

import org.apache.spark.{SparkConf, SparkContext}

/**
 * @author Hanpeng
 * @date 2021/1/14 14:39
 * @description: aggregateByKey 分区内和分区外的计算规则可以分别指定
 *
 */
object AggregateByKeyOperation {
  def main(args: Array[String]): Unit = {
    val sparkConf = new SparkConf().setMaster("local[*]").setAppName("Operator")
    val sc = new SparkContext(sparkConf)

    // TODO 算子 - (Key - Value类型)

    val rdd = sc.makeRDD(List(
      ("a", 1), ("a", 2),("a",100), ("a", 3), ("a", 4),("a",200)
    ),2)
    // (a,【1,2】), (a, 【3，4】)
    // (a, 2), (a, 4)
    // (a, 6)

    // aggregateByKey存在函数柯里化，有两个参数列表
    // 第一个参数列表,需要传递一个参数，表示为初始值
    //       主要用于当碰见第一个key的时候，和value进行分区内计算
    // 第二个参数列表需要传递2个参数
    //      第一个参数表示分区内计算规则
    //      第二个参数表示分区间计算规则

    // math.min(x, y)
    // math.max(x, y)
    //(x, y) => math.max(x, y), 分区内
    //(x, y) => x + y 分区间最大值的和
    /**
     * 从0开始 0和y比 返回两者之间的最大值(分区内最大值)
     * 对每个分区的数据的最大值,然后进行合并
     *
     */
    rdd.aggregateByKey(0)(
      (x, y) => math.max(x, y),
      (x, y) => x + y
    ).collect.foreach(println)
    sc.stop()
  }
}
